package cn.itcast.streaming

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object StreamingWordCount {

  def main(args: Array[String]): Unit = {
    // 1. 初始化环境
    // Spark Streaming 是基于Spark Core的
    // 在指定master的时候，不能指定一个线程（在Streaming运行的时候，需要开一个新的线程来一直监听数据的获取）
    val sparkConf = new SparkConf().setAppName("word count").setMaster("local[6]")
    val ssc = new StreamingContext(sparkConf, Seconds(5))  // 一秒一批次
    ssc.sparkContext.setLogLevel("WARN")
    // socketTextStream用于创建一个DStream，监听socket输入，当做文本来处理
    // sparkContext.textFile() 创建一个rdd
    // RDD -> Spark Core     DStream -> Spark Streaming（表示离散的流）
    // DStream可以理解为一个流式的RDD
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream(
      hostname = "192.168.88.100",
      port = 8888,
      storageLevel = StorageLevel.MEMORY_AND_DISK_SER
    )

    // 2. 数据的处理
    // 2.1 句子拆为单词
    val words: DStream[String] = lines.flatMap(_.split(" "))
    // 2.2 转换单词
    val tuples: DStream[(String, Int)] = words.map((_, 1))
    // 2.3 词频reduce
    val counts: DStream[(String, Int)] = tuples.reduceByKey(_ + _)

    // 3. 展示和启动
    counts.print()

    ssc.start()

    // main方法执行完毕后整个程序就会退出，需要阻塞主线程
    ssc.awaitTermination()

  }

}
